#Copyright (c) 2025, Alibaba Cloud and its affiliates;
#Licensed under the Apache License, Version 2.0 (the "License");
#you may not use this file except in compliance with the License.
#You may obtain a copy of the License at

#   http://www.apache.org/licenses/LICENSE-2.0

#Unless required by applicable law or agreed to in writing, software
#distributed under the License is distributed on an "AS IS" BASIS,
#WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#See the License for the specific language governing permissions and
#limitations under the License.



import json

import fce_util.libs.constants as constants
import fce_util.deploys.resource_mgr as resource
from fce_util.libs.log import g_logger
from fce_util.fceres import fce_res_common
from fce_util.libs.thread import MThread

logger = g_logger.logger

def cluster_create(args):
    try: 
        logger.debug('cluster_create enter,args:{}'.format(args))
        kube_config = args.kube_config if args.kube_config else constants.KUBE_CONFIG_DEFAULT_PATH
        namespace = args.namespace if args.namespace else constants.KUBE_NAMESPACE_DEFAULT
        cluster_version = args.cluster_version if args.cluster_version else constants.FCE_VERSION_DEFAULT

        #total resources check.TBD
        #params convert
        ray_cluster_cfg = {
            'cluster_name': args.cluster_name,
            'cluster_version': cluster_version,
            'cluster_id': args.cluster_id,
            'kube_config': kube_config,
            'namespace': namespace,
            #'operator_image': kube_config,
            'head_node': constants.head_node_config,
            'worker_node': constants.worker_node_config,
        }
        
        if args.head_node:
            head_node = json.loads(args.head_node)
            ray_cluster_cfg['head_node'] = head_node

        if args.worker_node:
            worker_node = json.loads(args.worker_node)
            ray_cluster_cfg['worker_node'] = worker_node
        
    except Exception as e:
        fce_res_common(success = False, errcode = 'InvalidParameters.ParseError', errmsg = e)
        return

    #crd configuration
    logger.info('ray cluster create start, ray_cluster_cfg: {}'.format(ray_cluster_cfg))
    status, result = resource.deploy_ray_cluster(ray_cluster_cfg)
    if status:
        resource.clean('cm', resource._getName(ray_cluster_cfg['cluster_id'], 'cm'), namespace)
        stdout = "deploy ray cluster: {} failed, error: {}".format(ray_cluster_cfg['cluster_name'], result)
        logger.error(stdout)
        fce_res_common(success = False, errcode = 'RayCluster.ClusterInstallFailed', errmsg = stdout)
        return
    
    try:
        thread = MThread(resource.checkServiceStatus, ray_cluster_cfg['cluster_name'], 'rc', namespace, 120, 1, kube_config)
        thread.start()
        thread.join()
        logger.debug('checkServiceStatus result is {}'.format(thread.result))
    except Exception as e:
        stdout = "check ray cluster {} status failed, error: {}".format(ray_cluster_cfg['cluster_name'], e)
        logger.error(stdout)
        resource.clean('rc', ray_cluster_cfg['cluster_name'], namespace)
        resource.clean('cm', resource._getName(ray_cluster_cfg['cluster_id'], 'cm'), namespace)
        fce_res_common(success = False, errcode = 'RayCluster.ClusterInstallFailed', errmsg = stdout)
        return

    
    #if not resource.checkServiceStatus(ray_cluster_cfg['cluster_name'], 'rc', namespace, 10, 3, kube_config):
    if not thread.result:
        stdout = "check ray cluster {} status failed".format(ray_cluster_cfg['cluster_name'])
        resource.clean('rc', ray_cluster_cfg['cluster_name'], namespace)
        resource.clean('cm', resource._getName(ray_cluster_cfg['cluster_id'], 'cm'), namespace)
        logger.error(stdout)
        fce_res_common(success = False, errcode = 'RayCluster.ClusterInstallFailed', errmsg = stdout)
        return

    msg = {'Success': True}
    msg['Message'] = {
        'Status': 'Running',
        'Msg': 'Kube ray cluster install success'
    }
    print(json.dumps(msg))

def FCE_cluster_create(args):
    if args.rayc == 'cluster_create':
        cluster_create(args)
        return
    args.this_parser.print_help()

def make(parser):
    """
    FCE Ray Cluster Create entry
    """
    parser.add_argument( 
        '--worker_node',
        dest='worker_node',
        nargs = '?',
        metavar = ('Worker_NODE'),
        help='Ray Cluster Worker Node Config',
    )
    
    parser.add_argument( 
        '--head_node',
        dest='head_node',
        nargs = '?',
        metavar = ('HEAD_NODE'),
        help='Ray Cluster Head Node Config',
    )
    
    parser.add_argument( 
        '--operator_image',
        dest='operator_image',
        nargs = '?',
        metavar = ('OPERATOR_IMAGE'),
        help='Kube Ray Operator Image Url',
    ) 

    parser.add_argument( 
        '--cluster_name',
        dest='cluster_name',
        nargs = '?',
        metavar = ('CLUSTER_NAME'),
        help='Ray Cluster Names',
    ) 
    
    parser.add_argument( 
        '--cluster_version',
        dest='cluster_version',
        nargs = '?',
        metavar = ('CLUSTER_VERSION'),
        help='Ray Cluster Version',
    )
    
    parser.add_argument( 
        '--cluster_id',
        dest='cluster_id',
        nargs = '?',
        metavar = ('CLUSTER_ID'),
        help='Ray Cluster Id',
    )  
    
    parser.add_argument( 
        '--namespace',
        dest='namespace',
        nargs = '?',
        metavar = ('NAMESPACE'),
        help='Ray Cluster Namespace',
    )
    
    parser.add_argument( 
        '--kube_config',
        dest='kube_config',
        nargs = '?',
        metavar = ('KUBE_CONFIG'),
        help='Kube Config File Path',
        required=False,
    )
    
    parser.set_defaults(
        func=FCE_cluster_create,
        this_parser=parser
    )